#pragma once

#include <brpc/server.h>
#include <butil/logging.h>

#include "etcd.hpp"
#include "logger.hpp"
#include "channel.hpp"
#include "mysql_chat_session_member.hpp"
#include "rabbitmq.hpp"
#include "utils.hpp"

#include "user.pb.h"
#include "base.pb.h"
#include "transmite.pb.h"

namespace xu
{
    class MsgTransmiteServicelmpl :public xu::MsgTransmitService
    {
    public:
        MsgTransmiteServicelmpl(const std::shared_ptr<odb::mysql::database>& mysql_client,
        const std::string &exchange_name,
        const std::string &routing_key,
        const MQClient::ptr &mq_client,
        const std::string &user_service_name,
        const ServiceManage::ptr &service_manage)
        :_mysql_session(std::make_shared<ManageSession>(mysql_client)),
        _exchange_name(exchange_name),
        _routing_key(routing_key),
        _mq_client(mq_client),
        _user_service_name(user_service_name),
        _service_manage(service_manage)
        {}
        ~MsgTransmiteServicelmpl(){}

        virtual void GetTransmitTarget(::google::protobuf::RpcController* controller,
                       const ::xu::NewMessageReq* request,
                       ::xu::GetTransmitTargetRsp* response,
                       ::google::protobuf::Closure* done)  override
        {
            brpc::ClosureGuard clo(done);
            response->set_request_id(request->request_id()); 
            auto err_rsp = [this,response](const std::string & err_msg)
            {
                response->set_success(false);
                response->set_errmsg(err_msg);
            };
            //1. 从请求中取出消息内容，会话 ID， 用户 ID
            std::string uid = request->user_id();
            std::string csid = request->chat_session_id();
            auto msg = request->message();
            
            //2. 根据用户 ID 从用户子服务获取当前发送者用户信息
            auto channel = _service_manage->Choose(_user_service_name);
            if(!channel)
            {
                LOG_ERROR("获取用户服务信道失败：{}", _user_service_name);
                err_rsp("获取用户服务信道失败");
                return;
            }

            xu::UserService_Stub stub(channel.get());
            xu::GetUserInfoReq req;
            xu::GetUserInfoRsp rsp;
            brpc::Controller cntl;
            std::string req_id = xu::uuid();
            req.set_request_id(req_id);
            req.set_user_id(uid);
            stub.GetUserInfo(&cntl,&req,&rsp,nullptr);
            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("获取用户服务失败：{}", uid);
                err_rsp("获取用户服务失败");
                return; 
            }
            //3. 根据消息内容构造完成的消息结构（分配消息 ID，填充发送者信息，填充消息产
            //生时间）
            MessageInfo msginfo;
            msginfo.set_message_id(uuid());
            msginfo.set_chat_session_id(csid);
            msginfo.set_timestamp(time(nullptr));
            msginfo.mutable_sender()->CopyFrom(rsp.user_info());
            msginfo.mutable_message()->CopyFrom(msg);
            //4. 将消息序列化后发布到 MQ 消息队列中，让消息存储子服务对消息进行持久化存储
            if(_mq_client->Publish(_exchange_name,_routing_key,msginfo.SerializeAsString()) == false)
            {
                LOG_ERROR("向消息队列添加失败：{} - {}",_exchange_name, _routing_key);
                err_rsp("向消息队列添加失败");
                return;   
            }
            //5. 从数据库获取目标会话所有成员 ID
            auto res = _mysql_session->SelectUserId(csid);
            //6. 组织响应（完整消息+目标用户 ID），发送给网关，告知网关该将消息发送给谁。
            response->set_success(true);
            response->mutable_message()->CopyFrom(msginfo);
            for(auto &e : res)
            {
                response->add_target_id_list(e);
            }

            LOG_INFO("持久化消息完成 {} ",uid);
        }
    private:
        ManageSession::ptr _mysql_session;  //对会话的管理

        std::string _exchange_name;         //交换机名称
        std::string _routing_key;
        MQClient::ptr _mq_client;           //消息的持久化

        std::string _user_service_name;     //用户服务的名称
        ServiceManage::ptr _service_manage; //服务发现
    };

    class TransmitServer
    {
    public:
        using ptr = std::shared_ptr<TransmitServer>;
        TransmitServer(const MQClient::ptr & mq_client,const ServiceManage::ptr & service_manage,
            const Discovery::ptr & discovery,const Registration::ptr &registration,
            const std::shared_ptr<odb::mysql::database> &mysql_client ,
            const  std::shared_ptr<brpc::Server> &server)
            :_mq_client(mq_client),
            _service_manage(service_manage),
            _discovery(discovery),
            _registration(registration),
            _mysql_client(mysql_client),
            _server(server)
        {}
        ~TransmitServer(){}
        void Start()
        {
            _server->RunUntilAskedToQuit();
        }
    private:
        MQClient::ptr _mq_client;                           //消息的持久化
        ServiceManage::ptr _service_manage;                 //服务发现信道管理
        Discovery::ptr _discovery;                          //服务发现
        Registration::ptr _registration;                    //etcd注册
        std::shared_ptr<odb::mysql::database> _mysql_client;//mysql客户端
        std::shared_ptr<brpc::Server> _server;              //rpc注册调用   
    };

    class TransmitServerBuilder
    {
    public:
        TransmitServerBuilder(){}
        ~TransmitServerBuilder(){}


        //构造服务发现信道
        void SetServiceManage(const std::string &service_name)
        {
            _user_service_name = service_name;
            //1.创建对象
            _service_manage = std::make_shared<xu::ServiceManage>();
            //2.添加发现服务
            _service_manage->AddFollow(service_name);
        }

        //构造etcd发现
        void SetDiscovery(const std::string &etcd_host,const std::string &base_service)
        {
            if(!_service_manage)
            {
                LOG_ERROR("ServiceManage未初始化");
                return ;
            }
            //1.绑定回调函数
            auto put_cb = std::bind(&xu::ServiceManage::onServiceOnline, _service_manage.get(), std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&xu::ServiceManage::onServiceOffline, _service_manage.get(), std::placeholders::_1, std::placeholders::_2);
            //2. 构造服务发现对象
           _discovery = std::make_shared<xu::Discovery>(etcd_host,base_service, put_cb, del_cb);
        }

        //构造etcd注册
        void SetRegistration(const std::string &host,const std::string &service_name,
            const std::string &access_host)
        {
            _registration = std::make_shared<Registration>(host);
            _registration->AddKV(service_name,access_host);
        }

        //构造Mysql客户端
        void SetMysqlClient(const std::string & user,
            const std::string &pswd,const std::string &db_name,
            const std::string &host,const std::string &cset)    
        {
            _mysql_client = MysqlClientFactory::Create(user,pswd,db_name,host,cset);
        }

        void SetRQClient(const std::string &user, const std::string &password, const std::string &host,
        const std::string &exchange_name, const std::string &queue_name, const std::string &routing_key)
        {
            _exchange_name = exchange_name;
            _routing_key = routing_key;
            _mq_client = std::make_shared<MQClient>(user,password,host);
            _mq_client->DeclareComponents(exchange_name,queue_name,routing_key);
        }

       
        void SetRpcServer(uint16_t port, int32_t timeout, uint8_t num_threads)
        {
            if(!_service_manage)
            {
                LOG_ERROR("ServiceManage未初始化");
                return ;
            }

            if(!_mysql_client)
            {
                LOG_ERROR("MysqlClient未初始化");
                return ;
            }

            if(! _mq_client)
            {
                LOG_ERROR("RQClient未初始化");
                return ;
            }
            
            _server = std::make_shared<brpc::Server>();
        //     const ManageSession::ptr& mysql_session,
        // const std::string &exchange_name,
        // const std::string &routing_key,
        // const MQClient::ptr &mq_client,
        // const std::string &user_service_name,
        // const ServiceManage::ptr &service_manage)

            MsgTransmiteServicelmpl *transmite_service = new  MsgTransmiteServicelmpl(
                _mysql_client,_exchange_name,_routing_key,_mq_client,_user_service_name,
                _service_manage);
            _server->AddService(transmite_service,brpc::ServiceOwnership::SERVER_OWNS_SERVICE);

            brpc::ServerOptions options;
            options.idle_timeout_sec = timeout;
            options.num_threads = num_threads;

            _server->Start(port, &options);
        }
        TransmitServer::ptr Build()
        {
           
            if(!_server)
            {
                LOG_ERROR("RpcServer未初始化");
                return TransmitServer::ptr();
            }
            // const MQClient::ptr & mq_client,const ServiceManage::ptr & service_manage,
            // const Discovery::ptr & discovery,const Registration::ptr &registration
            // const std::shared_ptr<odb::mysql::database> &mysql_client ,
            // const  std::shared_ptr<brpc::Server> &server)
            auto res = std::make_shared<TransmitServer>(_mq_client,_service_manage,
               _discovery, _registration,_mysql_client,_server);
            return res;
        }

    private:
        std::string _exchange_name;                         //交换机名称
        std::string _routing_key;
        MQClient::ptr _mq_client;                           //消息的持久化
        std::string _user_service_name;                     //用户服务的名称
        ServiceManage::ptr _service_manage;                 //服务发现信道管理
        Discovery::ptr _discovery;                          //服务发现
        Registration::ptr _registration;                    //etcd注册
        std::shared_ptr<odb::mysql::database> _mysql_client;//mysql客户端
        std::shared_ptr<brpc::Server> _server;              //rpc注册调用   
    };
}
